Summary

‘ipayipi’ was built for processing time-series data using ‘data.table’ functionality and speed for processing large datasets. This vignette introduces batch processing of time-series data using R ipayipi: covering extraction and standardisation of logger data. Specifically, the five steps outlined in this vignette are:

1. Load R ipayipi (install from GitHub),
2. Initiate pipeline,
3. Imbibe data,
4. Standardise data, &
5. Make station files.

An overview of the code to execute these five steps are shown below. This code is all that is required to run each time new data is available. For more details and options for each step, see respective function documentation/help files.

## 1. load pacakge
library(ipayipi)

## 2. initiate pipeline ----
# define our general working directory: wherever data is going to be processed ...
wd <- "ipayipi/data-raw/eg_data/met_eg"

# define directory from which data will be sourced
sd <- "ipayipi/data-raw/eg_data/met_eg/dta_in"

# setup pipeline directory structure
pipe_house <- ipip_init(work_dir = wd, source_dir = sd)

## 3. read & imbibe in data ----
logger_data_import_batch(pipe_house)
imbibe_raw_batch(pipe_house, file_ext_in = ".dat",
  data_setup = ipayipi::cs_toa5)

## 4. standardise data ----
header_sts(pipe_house)
phenomena_sts(pipe_house)

## store standardised data ----
# transfer standardised data to the nomvet_room
transfer_sts_files(pipe_house)

## 5. append standardised data files ----
append_station_batch(pipe_house)

Introduction

Organising time-series data from loggers, such as, weather stations or ground-water sensors, requires no small amount of data structuring. ipayipi helps make this process dynamic and structured, so that, processing, down (and back up) a pipeline, is traceable. Importantly, ipayipi preserves raw-data integrity and initiates archival—so raw data is available in a standardised format—and the processing of this data can be updated.

The package can be found on GitHub and can be installed using the ‘devtools’ package: devtools::install_github("SAEONData/ipayipi").

Initiate pipeline: the ‘pipe_house’

The ‘pipe_house’ is the name of the directory where we will initiate a data-pipeline structure. To keep things simple, we will only use our ‘pipe_house’ for defined data streams. In this vignette the data stream consists of meteorological data gathered from a weather station data logger. ipayipi handles most time-series data formats readable from flat files into R. Data for this vignette can be downloaded from GitHub in the package’s raw-data folder here.

# setting up the 'pipe_house'
## general pipeline working directory
wd <- "ipayipi/data-raw/eg_data/met_eg"

## data souce folder
sd <- "ipayipi/data-raw/eg_data/met_eg/dta_in"

## initiate pipeline
pipe_house <- ipip_init(work_dir = wd, source_dir = sd)
print(pipe_house)

## $wait_room
##                                   wait_room 
## "ipayipi/data-raw/eg_data/met_eg/wait_room" 
## 
## $nomvet_room
##                                   nomvet_room 
## "ipayipi/data-raw/eg_data/met_eg/nomvet_room" 
## 
## $ipip_room
##                                   ipip_room 
## "ipayipi/data-raw/eg_data/met_eg/ipip_room" 
## 
## $raw_room
##                                   raw_room 
## "ipayipi/data-raw/eg_data/met_eg/raw_room" 
## 
## $source_dir
## [1] "ipayipi/data-raw/eg_data/met_eg/dta_in"

What has ipip_init() done? It has created the following directories, if they don’t already exist*:

1. ‘source_dir’: where new logger data is going to be made available.
2. ‘wait_room’: waiting room for imbibing data into the pipeline.
3. ‘nomvet_room’: where standardised/corrected logger files get archived (nomenclature vetted).
4. ‘ipip_room’: here data get appended into contiguous single station records, and processed.
5. ‘raw_room’: where ‘unaltered’ raw data gets pushed.

*NB! Running this function will not overwrite existing data.

Imbibe data

In this step, data gets pulled from pipelines data source, that is, the ‘source directory’ (pipe_house$source_dir), into the ‘waiting room’ (pipe_house$wait_room). The example data contains two-years of Cambell Scientific logger text files derived from sensors on a SAEON meteorological station in northern Maputaland, South Africa.

# copy data from source to the wait_room
logger_data_import_batch(pipe_house = pipe_house,
  file_ext = ".dat", # the file extension (with period) of raw data files
  verbose = FALSE, # set to TRUE to report progress in the terminal
  unwanted = "02.2022" # excluding the import of Feb 2022 to make data 'gap'
)

Now that some data is in the ‘wait_room’ directory we can read it into R. Note the pre-set ‘data_setup’ option for Cambell Scientific TOA5 formatted files ipayipi::cs_toa5.

imbibe_raw_batch(pipe_house = pipe_house,
  data_setup = ipayipi::cs_toa5, # standard for reading t0a5 formatted files 
  record_interval_type = "continuous"
)

For more on data input formats, that is, the ‘data_setup’ argument, see the help files of the imbibe_raw_logger_dt() function (i.e., ?imbibe_raw_logger_dt).

Record-interval type is an important parameter. ipayipi handles continuous, event-based (discontinuous), and mixed time-series data types. Record intervals get evaluated using the record_interval_eval() function. Record interval information will be important for further steps, such as, identifying ‘gaps’ or missing data automatically.

Standardise data

Both file-header information, plus other phenomena (variable) metadata, will now be standardised. The spelling/synonyms of file names and associated header metadata have to be scrutinised first. Only after header information gets standardised, can we move on to working on the phenomena. These steps are essential for automating file record appending and downstream data correction/processing (e.g., drift correction).

header_sts(pipe_house)

If it is the first-time running header_sts(), or new synonyms get introduced into pipe-house directory, header_sts() will produce a warning. This is because the user needs to define new nomenclature standards. Unstandardised names (or columns) have the preffix ‘uz’. These standards get stored in a file called ‘nomtab.rns’ in the ‘waiting room’. If this file is deleted—a new one will be generated—but the user will have to populate the tables with synonym vocab.

The nomenclature table in the ‘waiting room’ can be updated from ‘csv’ format (or directly in R). If a new synonym gets introduced—the file containing new nomenclature will be skipped in further processing—a ‘csv’ version of the ‘nomtab.rns’ will be copied to the ‘waiting room’ for editing.

Only the following fields — with NAs — require editing in the ‘nomtab’ ‘csv’:

• location: standardised location shorthand,
• station: standardised station name,
• standard title: station name. Generally a concatenation of location and station.
• record interval type: Either ‘continuous’, ‘mixed’ or ‘event_based’. Automatically generated and saved as unstandardised (‘uz’) using record_interval_eval(). However, if there are in sufficient data records, e.g., one record in a continuous data flow, this needs manual checking.
• record interval : In addition to ‘record interval type’ the ‘record interval’ describes the nature of time series data. The ‘record interval’ is a standardised string describing the interval between data records, e.g., “1_days”, for daily. Can be “discnt” for discontinuous data. For mixed data the ‘record interval’ describes the continuous interval. The ‘record interval’ string can be standardised using ipayipi::record_interval_eval().
• table name: shorthand description of the data table from respective logger. This should generally be a description of the time interval between recordings, or for discontinuous, say rainfall data, be ‘raw_rain’, for example.

pt <- read.csv("ipayipi/data-raw/eg_data/met_eg/nomtab_display.csv")
kbl(pt) |>
  kable_paper("hover") |>
  kable_styling(font_size = 11) |>
  column_spec(c(1, 7:8, 11), background = "#df9a86") |>
  column_spec(c(2:6, 9:10, 12), background = "#aed8f0") |>
  scroll_box(width = "100%", height = "400px")

Once NA values of the above fields have been populated the edited ‘csv’ will be imbibed into the pipeline structure when rerunning header_sts(pipe_house)—this function will imbibe the most recently updated ‘csv’ nomenclature table from the ‘wait_room’ into the pipeline, and standardised header nomenclature.

In step with good tidy data standards, keep nomenclature to ‘snake case’ with no special characters (bar the useful underscore’). ’ Standardising phenomena metadata follows a similar process as for header-data standardisation. If the phenomena standards have been described and there is a ‘phentab.rps’ in the ‘waiting room’, running the below code updates all files phenomena details.

phenomena_sts(pipe_house = pipe_house, unwanted = "display")

If there is no ‘phenomena table’ (‘phentab.rps’), one NA values in the ‘csv’ copy need to be described. The following fields in the ‘csv’ phentab must be populated:

• phen_name_full: A descriptive name of the phenomenon (variable).
• phen_type: the type of phenomena (e.g., atmospheric pressure).
• phen_name: the name (used as a column header) of the phenomena. Must be created to avoid duplications.
• units: the measurement unit, e.g., bars.
• measure: describes how sampling has been done over/at the sampling interval, e.g., mean, min, sample, etc.

Additional fields that are not mandatory include:

• f_convert: A numeric multiplier that will be applied to this phenomena. Useful for converting from unstandardised to standardised units.

If an ‘f_convert’ factor (scroll right on the table below) is applied to phenomena, the standardised units must be different from the unstandardised units (uz_units) in the phenomena table. This ensures that phenomena that are appended have similar units.

After filling in details, to replace NA values, rerun phenomena_sts(pipe_house), to imbibe the updated phenomena descriptions, and update the logger data being standardised.

Standardised data files get transferred to the ‘nomenclature vetted’ directory (‘nomtab room’) using the function below. After being transferred, files in the waiting room (except the nomtab and phentab standards) are automatically removed.

# move standardised files to a storage directory
transfer_sts_files(pipe_house)

Archiving raw data files: Before removing raw unstandardised files—if there is a ‘raw_room’ directory in the pipeline working directory—raw input data files will be copied to this directory and filed in folders by year and month of the lasted date of recording. This is done by the imbibe_raw_batch() function.

Make station files

The append_station_batch() function updates station files in the ‘ipip_room’ with files from the ‘nomvet_room’.

# append station files + metadata records
# note the 'cores' argument --- parallel processing supported on Linux systems
append_station_batch(pipe_house, cores = 2)

Now that a station file has been generated for the Vasi Science Centre weather station we can check what tables have been created/appended. Station files are maintained in the ‘ipip_room’ of the pipeline’s folder structure.

# list station files in the ipip directory
sf <- dta_list(
  input_dir = pipe_house$ipip_room, # search directory
  file_ext = ".ipip", # note the station's default file extension
)

# check what stations are in the ipip room
print(sf)

## [1] "mcp_vasi_science_centre_aws.ipip"

# read in the station file
sf <- readRDS(file.path(pipe_house$ipip_room, sf[1]))

# names of the tables stored in the station file
names(sf)

## [1] "data_summary"      "gaps"              "logg_interfere"    "phen_data_summary" "phens"             "raw_1_days"        "raw_1_hours"       "raw_5_mins"

# a look 1st data row of 5 minute data
print(sf$raw_5_min[1, ])

##    id           date_time wind_speed wind_dir wind_dir_sd temp_air_avg humid_rel_pcnt solar_rad_avg uv_rad_avg rain_tot temp_ground_avg
## 1:  0 2020-06-11 12:35:00       3.29    309.2       42.72        31.27          38.04         584.1      24.86        0           32.92

# use kableExtra to view first 20 data rows --- default printing of tables doesn't
#  looks good in html
kbl(sf$raw_5_min[1:20, ]) |>
  kable_paper("hover") |>
  kable_styling(font_size = 11) |>
  scroll_box(width = "100%", height = "400px")

# using kableExtra
kbl(sf$data_summary) |> kable_paper("hover") |> kable_styling(font_size = 11) |>
  scroll_box(width = "100%", height = "400px")

# using kableExtra
kbl(sf$phens) |> kable_paper("hover") |> kable_styling(font_size = 11) |>
  scroll_box(width = "100%", height = "400px")

# using kableExtra
kbl(sf$phen_data_summary) |> kable_paper("hover") |> kable_styling(font_size = 11) |>
  scroll_box(width = "100%", height = "400px")

gap_eval_batch(pipe_house, cores = 3)
# read in the station file
sf <- readRDS(file.path("ipayipi/data-raw/eg_data/met_eg",
  "ipip_room/mcp_vasi_science_centre_aws.ipip"))
sf$gaps

##    gid euid gap_type   phen  table_name           gap_start             gap_end    dt_diff_s gap_problem_thresh_s problem_gap notes
## 1:  13   NA     auto logger  raw_1_days 2022-01-12 00:00:00 2022-02-10 00:00:00 2505600 secs                86400        TRUE  <NA>
## 2:  13   NA     auto logger raw_1_hours 2022-01-12 11:00:00 2022-02-09 17:00:00 2440800 secs                 3600        TRUE  <NA>
## 3:  13   NA     auto logger  raw_5_mins 2022-01-12 11:55:00 2022-02-09 16:10:00 2434500 secs                  300        TRUE  <NA>

p <- plot_raw_availability(pipe_house)
p <- p$plot_availability + scale_colour_sunset(discrete = TRUE) +
      labs(color = "Station") +
      theme(legend.position = "none")
plotly::ggplotly(p)